/*-------------------------------------------------------------------------
 *
 * pg_lsn.c
 *	  Operations for the pg_lsn datatype.
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/utils/adt/pg_lsn.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "funcapi.h"
#include "libpq/pqformat.h"
#include "utils/builtins.h"
#include "utils/numeric.h"
#include "utils/pg_lsn.h"

#define MAXPG_LSNLEN			17
#define MAXPG_LSNCOMPONENT	8

/*----------------------------------------------------------
 * 格式化和转换例程。
 *---------------------------------------------------------*/

XLogRecPtr pg_lsn_in_internal(const char *fc_str, bool *fc_have_error)
{
	int			fc_len1,
				fc_len2;
	uint32		fc_id,
				fc_off;
	XLogRecPtr	fc_result;

	Assert(fc_have_error != NULL);
	*fc_have_error = false;

	/* 输入格式的合理性检查。 */
	fc_len1 = strspn(fc_str, "0123456789abcdefABCDEF");
	if (fc_len1 < 1 || fc_len1 > MAXPG_LSNCOMPONENT || fc_str[fc_len1] != '/')
	{
		*fc_have_error = true;
		return InvalidXLogRecPtr;
	}
	fc_len2 = strspn(fc_str + fc_len1 + 1, "0123456789abcdefABCDEF");
	if (fc_len2 < 1 || fc_len2 > MAXPG_LSNCOMPONENT || fc_str[fc_len1 + 1 + fc_len2] != '\0')
	{
		*fc_have_error = true;
		return InvalidXLogRecPtr;
	}

	/* 解码结果。 */
	fc_id = (uint32) strtoul(fc_str, NULL, 16);
	fc_off = (uint32) strtoul(fc_str + fc_len1 + 1, NULL, 16);
	fc_result = ((uint64) fc_id << 32) | fc_off;

	return fc_result;
}

Datum pg_lsn_in(PG_FUNCTION_ARGS)
{
	char	   *fc_str = PG_GETARG_CSTRING(0);
	XLogRecPtr	fc_result;
	bool		fc_have_error = false;

	fc_result = pg_lsn_in_internal(fc_str, &fc_have_error);
	if (fc_have_error)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input syntax for type %s: \"%s\"",
						"pg_lsn", fc_str)));

	PG_RETURN_LSN(fc_result);
}

Datum pg_lsn_out(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn = PG_GETARG_LSN(0);
	char		fc_buf[MAXPG_LSNLEN + 1];
	char	   *fc_result;

	snprintf(fc_buf, sizeof fc_buf, "%X/%X", LSN_FORMAT_ARGS(fc_lsn));
	fc_result = pstrdup(fc_buf);
	PG_RETURN_CSTRING(fc_result);
}

Datum pg_lsn_recv(PG_FUNCTION_ARGS)
{
	StringInfo	fc_buf = (StringInfo) PG_GETARG_POINTER(0);
	XLogRecPtr	fc_result;

	fc_result = pq_getmsgint64(fc_buf);
	PG_RETURN_LSN(fc_result);
}

Datum pg_lsn_send(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn = PG_GETARG_LSN(0);
	StringInfoData fc_buf;

	pq_begintypsend(&fc_buf);
	pq_sendint64(&fc_buf, fc_lsn);
	PG_RETURN_BYTEA_P(pq_endtypsend(&fc_buf));
}


/*----------------------------------------------------------
 *	PostgreSQL LSNs 的操作符
 *---------------------------------------------------------*/

Datum pg_lsn_eq(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 == fc_lsn2);
}

Datum pg_lsn_ne(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 != fc_lsn2);
}

Datum pg_lsn_lt(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 < fc_lsn2);
}

Datum pg_lsn_gt(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 > fc_lsn2);
}

Datum pg_lsn_le(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 <= fc_lsn2);
}

Datum pg_lsn_ge(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_BOOL(fc_lsn1 >= fc_lsn2);
}

Datum pg_lsn_larger(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_LSN((fc_lsn1 > fc_lsn2) ? fc_lsn1 : fc_lsn2);
}

Datum pg_lsn_smaller(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);

	PG_RETURN_LSN((fc_lsn1 < fc_lsn2) ? fc_lsn1 : fc_lsn2);
}

/* btree 索引操作类支持 */
Datum pg_lsn_cmp(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_a = PG_GETARG_LSN(0);
	XLogRecPtr	fc_b = PG_GETARG_LSN(1);

	if (fc_a > fc_b)
		PG_RETURN_INT32(1);
	else if (fc_a == fc_b)
		PG_RETURN_INT32(0);
	else
		PG_RETURN_INT32(-1);
}

/* hash 索引操作类支持 */
Datum pg_lsn_hash(PG_FUNCTION_ARGS)
{
	/* 我们可以直接使用 hashint8 */
	return hashint8(fcinfo);
}

Datum pg_lsn_hash_extended(PG_FUNCTION_ARGS)
{
	return hashint8extended(fcinfo);
}


/*----------------------------------------------------------
 *	对 PostgreSQL LSNs 的算术操作符。
 *---------------------------------------------------------*/

Datum pg_lsn_mi(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn1 = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsn2 = PG_GETARG_LSN(1);
	char		fc_buf[256];
	Datum		fc_result;

	/* 输出的范围可以达到正负 2^63 - 1。 */
	if (fc_lsn1 < fc_lsn2)
		snprintf(fc_buf, sizeof fc_buf, "-" UINT64_FORMAT, fc_lsn2 - fc_lsn1);
	else
		snprintf(fc_buf, sizeof fc_buf, UINT64_FORMAT, fc_lsn1 - fc_lsn2);

	/* 转换为数字。 */
	fc_result = DirectFunctionCall3(numeric_in,
								 CStringGetDatum(fc_buf),
								 ObjectIdGetDatum(0),
								 Int32GetDatum(-1));

	return fc_result;
}

/*
 * 将字节数加到 pg_lsn，得到新的 pg_lsn。
 * 必须处理正负字节数的情况。
 */
Datum pg_lsn_pli(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn = PG_GETARG_LSN(0);
	Numeric		fc_nbytes = PG_GETARG_NUMERIC(1);
	Datum		fc_num;
	Datum		fc_res;
	char		fc_buf[32];

	if (numeric_is_nan(fc_nbytes))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot add NaN to pg_lsn")));

	/* 转换为 numeric */
	snprintf(fc_buf, sizeof(fc_buf), UINT64_FORMAT, fc_lsn);
	fc_num = DirectFunctionCall3(numeric_in,
							  CStringGetDatum(fc_buf),
							  ObjectIdGetDatum(0),
							  Int32GetDatum(-1));

	/* 相加两个 numeric */
	fc_res = DirectFunctionCall2(numeric_add,
							  NumericGetDatum(fc_num),
							  NumericGetDatum(fc_nbytes));

	/* 转换为 pg_lsn */
	return DirectFunctionCall1(numeric_pg_lsn, fc_res);
}

/*
 * 从 pg_lsn 减去字节数，得到新的 pg_lsn。
 * 必须处理正负字节数的情况。
 */
Datum pg_lsn_mii(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_lsn = PG_GETARG_LSN(0);
	Numeric		fc_nbytes = PG_GETARG_NUMERIC(1);
	Datum		fc_num;
	Datum		fc_res;
	char		fc_buf[32];

	if (numeric_is_nan(fc_nbytes))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot subtract NaN from pg_lsn")));

	/* 转换为 numeric */
	snprintf(fc_buf, sizeof(fc_buf), UINT64_FORMAT, fc_lsn);
	fc_num = DirectFunctionCall3(numeric_in,
							  CStringGetDatum(fc_buf),
							  ObjectIdGetDatum(0),
							  Int32GetDatum(-1));

	/* 相减两个 numeric */
	fc_res = DirectFunctionCall2(numeric_sub,
							  NumericGetDatum(fc_num),
							  NumericGetDatum(fc_nbytes));

	/* 转换为 pg_lsn */
	return DirectFunctionCall1(numeric_pg_lsn, fc_res);
}
